#include "TcpConnection.h"
#include "Logger.h"
#include "Channel.h"
#include "Socket.h"
#include "EventLoop.h"

#include <functional>
#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <strings.h>
#include <netinet/tcp.h>
#include <string>

/**
 * @brief 检查loop
 *
 * @param loop
 * @return EventLoop*
 */
static EventLoop *CheckLoopNotNull(EventLoop *loop)
{
    if (loop == nullptr)
    {
        LOG_FATAL("%s:%s:%d TcpConnection Loop is null!", __FILE__, __FUNCTION__, __LINE__);
    }
    return loop;
}
/**
 * @brief Construct a new Tcp Connection:: Tcp Connection object
 *
 * @param loop
 * @param name
 * @param sockfd
 * @param localAddr
 * @param peerAddr
 */
TcpConnection::TcpConnection(EventLoop *loop, const std::string &nameArg, int sockfd, const InetAddress &localAddr, const InetAddress &peerAddr)
    : loop_(CheckLoopNotNull(loop)), name_(nameArg), state_(kConnecting), reading_(true), socket_(new Socket(sockfd)),
      channel_(new Channel(loop, sockfd)), localAddr_(localAddr), peerAddr_(peerAddr), highWaterMark_(64 * 1024 * 1024)
{
    // 注册event回调函数
    channel_->setReadCallback(std::bind(&TcpConnection::handleRead, this, std::placeholders::_1));
    channel_->setWriteCallback(std::bind(&TcpConnection::handleWrite, this));
    channel_->setCloseCallback(std::bind(&TcpConnection::handleClose, this));
    channel_->setErrorCallback(std::bind(&TcpConnection::handleError, this));

    LOG_INFO("TcpConnection::Construct[%s] at fd=%d\n", name_.c_str(), sockfd);
    socket_->setKeepAlive(true);
}
/**
 * @brief Destroy the Tcp Connection:: Tcp Connection object
 *
 */
TcpConnection::~TcpConnection()
{
    LOG_INFO("TcpConnection::Destory[%s] at fd=%d state=%d\n", name_.c_str(), channel_->fd(), (int)state_);
}
/**
 * @brief 提供给用户-发送数据
 *
 * @param buff
 */
void TcpConnection::send(const std::string &buff)
{
    if (state_ == kConnected)
    {
        if (loop_->isInLoopThread())
        {
            sendInLoop(buff.c_str(), buff.size());
        }
        else // 唤醒阻塞中的subloop，再执行回调
        {
            loop_->runInLoop(std::bind(&TcpConnection::sendInLoop, this, buff.c_str(), buff.size()));
        }
    }
}
void TcpConnection::sendInLoop(const void *message, size_t len)
{
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;
    // 链接断开
    if (state_ == kDisconnected)
    {
        LOG_ERROR("disconnected give up writing!");
        return;
    }
    // 第一次写事件&&缓冲区没有待发送数据
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    {
        nwrote = ::write(channel_->fd(), message, len);
        if (nwrote >= 0)
        {
            remaining = len - nwrote;
            if (remaining == 0 && writeCompleteCallback_) // 全部发送完毕
            {
                loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this())); // 写完全回调
            }
        }
        else
        {
            nwrote = 0;
            if (errno != EWOULDBLOCK) // 不是写入失败
            {
                LOG_ERROR("TcpConnection::sendInLoop error");
                if (errno == EPIPE || errno == ECONNRESET) // SIGPIPE||RESET
                {
                    faultError = true;
                }
            }
        }
    }
    // 没有发送完数据||之前有剩余数据
    if (!faultError && remaining > 0)
    {
        // 目前发送缓冲区剩余的待发送数据的长度
        size_t oldLen = outputBuffer_.readableBytes();
        if (oldLen + remaining >= highWaterMark_ && oldLen < highWaterMark_ && highWaterMarkCallback_) // 发送超过水位线
        {
            loop_->queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining)); // 高水位回调
        }
        // 将当前数据缓存，并关注对应事件
        outputBuffer_.append((char *)message + nwrote, remaining);
        if (!channel_->isWriting())
        {
            channel_->enableWriting();
        }
    }
}
/**
 * @brief 关闭
 *
 */
void TcpConnection::shutdown()
{
    if (state_ == kConnected)
    {
        setState(kDisconnecting);
        loop_->runInLoop(
            std::bind(&TcpConnection::shutdownInLoop, this));
    }
}
void TcpConnection::shutdownInLoop()
{
    if (!channel_->isWriting()) // 数据发送完成
    {
        socket_->shutdownWrite(); // 关闭写端
        LOG_INFO("TcpConnection shutdownWrite fd=%d\n", socket_->fd());
    }
}
/**
 * @brief 链接建立
 *
 */
void TcpConnection::connectEstablished()
{
    // 状态更改和注册
    setState(kConnected);
    channel_->tie(shared_from_this());       // 绑定weak_ptr，便于观察上层TcpConnection是否释放
    channel_->enableReading();               // 注册读事件
    connectionCallback_(shared_from_this()); // 链接建立回调
    LOG_INFO("fd=%d connection established at %p\n", channel_->fd(), loop_);
}
/**
 * @brief 链接销毁
 *
 */
void TcpConnection::connectDestoryed()
{
    // 更改状态和channel移除
    if (state_ == kConnected)
    {
        setState(kDisconnected);
        channel_->disableAll();
        connectionCallback_(shared_from_this());
    }
    channel_->remove();
}
/**
 * @brief 处理event读事件
 *
 * @param receiveTime
 */
void TcpConnection::handleRead(Timestamp receiveTime)
{
    int saveErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &saveErrno);
    if (n > 0)
    {
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    }
    else if (n == 0) // 没有数据却有EPOLLIN-对端关闭
    {
        handleClose();
    }
    else
    {
        errno = saveErrno;
        LOG_ERROR("TcpConnection::handleRead error");
        handleError();
    }
}
/**
 * @brief 处理event写事件
 *
 */
void TcpConnection::handleWrite()
{
    if (channel_->isWriting())
    {
        int saveErrno = 0;
        ssize_t n = outputBuffer_.writeFd(channel_->fd(), &saveErrno);
        if (n > 0)
        {
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0) // 发送完毕
            {
                channel_->disableWriting();
                if (writeCompleteCallback_)
                {
                    loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
                }
                if (state_ == kDisconnecting) // 正在关闭
                {
                    shutdownInLoop();
                }
            }
        }
        else
        {
            LOG_ERROR("TcpConnection::handleWrite error");
        }
    }
    else
    {
        LOG_ERROR("TcpConnecion fd=%d is down, no more writing\n", channel_->fd());
    }
}
/**
 * @brief 处理关闭事件
 *
 */
void TcpConnection::handleClose()
{
    LOG_INFO("TcpConnection::handleClose() fd=%d state=%d\n", channel_->fd(), (int)state_);
    setState(kDisconnected);
    channel_->disableAll();

    TcpConnectionPtr connPtr(shared_from_this());
    connectionCallback_(connPtr);
    closeCallback_(connPtr);
}
void TcpConnection::handleError()
{
    int optval;
    socklen_t optlen = sizeof(optval);
    int err = 0;
    if (::getsockopt(channel_->fd(), SOL_SOCKET, SO_ERROR, &optval, &optlen) < 0) // 获取错误信息
    {
        err = errno;
    }
    else
    {
        err = optval;
    }
    LOG_ERROR("TcpConnecion::handleError name:%s SO_ERROR:%d", name_.c_str(), err);
}
